
package node;

import java.io.Serializable;
import java.rmi.AccessException;
import java.rmi.AlreadyBoundException;
import java.rmi.NotBoundException;
import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

import log.Logger;
import messaging.InterNodeMessage;
import messaging.RemoveNodeMessage;
import messaging.TokenNodeMessage;
import messaging.TokenType;
import rebalance.HandleRebalanceMessages;
import rebalance.RebalanceChecks;
import remote.FutureMessage;
import remote.RemoteDistributedHashMap;
import remote.UnicastRemoteDistributedHashMap;

/**
 * 
 * @author Team 4
 * @version Spring 2013 This class is the thread that will poll all the channels
 *          of each node and looks for messages on the queue.
 * 
 */
public class QueueProcessor implements Runnable
{
  /**
   * Put key.
   */
  private int PUT_KEY = 1;
  /**
   * Get key.
   */
  private int GET_KEY = 2;
  /**
   * remove key.
   */
  private int REMOVE_KEY = 3;
  /**
   * Automatic rebalance trigger key.
   */
  private int REBALANCE = 4;
  /**
   * the key for add node message.
   */
  private int ADD_NODE = 6;
  /**
   * remove node.
   */
  private int REMOVE_NODE = 5;
  /**
   * token.
   */
  private int TOKEN = 7;
  /**
   * multiple store method id.
   */
  private int MULTIPLESTORE = 8;
  /**
   * rebalance update key.
   */
  private int FAULT_TOLERANCE = 9;
  /**
   * Wait time to poll the queues.defaults to 3 seconds.
   */
  private int WAIT_THREAD = 50;
  /**
   * wait time to send message to the rebalanced node.
   */
  private int WAIT_MESSAGE = 1000;
  /**
   * the node id
   */
  public int my_node_id;
  /**
   * list of queues this node needs to process.
   */
  public ArrayList<BlockingQueue<InterNodeMessage>> queues;
  /**
   * Boolean to run the thread.
   */
  private boolean threadRun = true;
  /**
   * NodeManager reference
   */
  public NodeManager nodeman;

  /**
   * Constructor
   */
  public QueueProcessor(int node_id, NodeManager nm)
  {
    this.my_node_id = node_id;
    this.nodeman = nm;
    queues = nm.node_channel.get(my_node_id).getBlockingQueue();
  }

  /**
   * Setter to set the boolean to run thread.
   * 
   * @param set_bool - boolean to set running the thread.
   */
  public void setRun(final boolean set_bool)
  {
    this.threadRun = set_bool;
  }

  /**
   * This is the main run method, runs the thread , thread polls all the
   * channels to which the node is listening to.If a message is in any queue, it
   * takes it to process them.
   */
  @Override
  public void run()
  {
    InterNodeMessage take = null;
    while (threadRun)
    {
      for (int i = 0; i < queues.size(); i++)
      {
        try
        {
          take =
              (InterNodeMessage) nodeman.node_channel.get(my_node_id).getBlockingQueue()
                  .get(i).poll(WAIT_THREAD, TimeUnit.MILLISECONDS);
          if (take != null)
          {
            // process the add node message
            if (take.getMethod_name() == ADD_NODE)
            {
              Logger.out(" A add node message received for node " + this.my_node_id);
              final QueueHelper queue_help = new QueueHelper(this.nodeman);
              queue_help.processAddNodeMessage(take, this.my_node_id);
            }
            // process the remove node message
            if (take.getMethod_name() == REMOVE_NODE)
            {
              Logger.out(" A remove node message received for node " + this.my_node_id);
              Logger
                  .writeContent(nodeman.LOGGER_FILE_NAME,
                                " a remove node message receive for node " + this.my_node_id);
              this.processMessagenode(take);
            }
            if (take.getMethod_name() == PUT_KEY || take.getMethod_name() == GET_KEY ||
                take.getMethod_name() == REMOVE_KEY || take.getMethod_name() == REBALANCE ||
                take.getMethod_name() == MULTIPLESTORE ||
                take.getMethod_name() == FAULT_TOLERANCE)
            {
              this.routingMessages(take);
            }
            if (take.getMethod_name() == TOKEN)
            {
              final TokenNodeMessage token_message =
                  new TokenNodeMessage(take.getMyTokenPacket(), this.my_node_id, this.nodeman);
              token_message.processMessage();
              if (token_message.isMySendAcross())
              {
                take.setMyTokenPacket(token_message.getTokenPacket());
                take.setMy_source_node(this.my_node_id);
                take.setMy_originalsource_node(this.my_node_id);

                int dest_node_id = my_node_id + 1;
                take.setMy_originaldest_node(dest_node_id);
                take.setMy_dest_node(dest_node_id);
                if (dest_node_id == NodeManager.total_nodes_chord)
                {
                  dest_node_id = 0;
                }
                Logger.writeContent("LOG-"+this.my_node_id  + ".txt", "Token data updated and sending to next node");
                String dest_node_str = nodeman.rmi_NodeFind(dest_node_id);
                // Logger.out(message.toString());
                final RemoteDistributedHashMap to_send =
                    (RemoteDistributedHashMap) NodeManager.registry.lookup(dest_node_str);
                to_send.send(take);
              }
              else
              {

                // System.out.println("This is my node message and need to send to client "
                // + token_message.getTokenPacket().getMyFirstKey());

                // System.out.println("This is myy node message and need to send to client "
                // + token_message.getTokenPacket().getMyNodeCount());
                Serializable val = null;
                if (token_message.getTokenPacket().getMyTokenType() == TokenType.COUNT)
                {
                  val = token_message.getTokenPacket().getMyKeyCount();
                }
                else if (token_message.getTokenPacket().getMyTokenType() == TokenType.ACTUALCOUNT)
                {
                  Iterator<Integer> temp =
                      token_message.getTokenPacket().getMyActualCount().iterator();
                  StringBuffer sb = new StringBuffer();
                  while (temp.hasNext())
                  {
                    sb.append(temp.next().toString());
                    sb.append("A, ");
                  }
                  val = token_message.getTokenPacket().getMyActualCount();
                }
                else if (token_message.getTokenPacket().getMyTokenType() == TokenType.FIRSTKEY)
                {
                  val = token_message.getTokenPacket().getMyFirstKey();
                }
                else if (token_message.getTokenPacket().getMyTokenType() == TokenType.LASTKEY)
                {
                  val = token_message.getTokenPacket().getMyLastKey();
                  take.getHashTableMessage().setvalue(val);
                  final FutureMessage mes =
                      nodeman.node_stubs.get(this.my_node_id).my_futures.get(take
                          .getHashTableMessage().getUniqueId());
                  mes.set(take.getHashTableMessage().getvalue());
                }
                take.getHashTableMessage().setvalue(val);
                final FutureMessage mes =
                    nodeman.node_stubs.get(this.my_node_id).my_futures.get(take
                        .getHashTableMessage().getUniqueId());
                Logger.writeContent("LOG-"+this.my_node_id  + ".txt",
                                    "Token data updated by all nodes and sending back to client");
                mes.set(take.getHashTableMessage().getvalue());

              }
            }
          }
        }
        catch (final InterruptedException e)
        {
          Logger.writeContent(nodeman.LOGGER_FILE_NAME, " interrupted in thread " + e);
          System.exit(1);
        }
        catch (final AlreadyBoundException e)
        {
          Logger.writeContent(nodeman.LOGGER_FILE_NAME, " already bound exception in thread " +
                                                        e);
          System.exit(1);
        }
        catch (AccessException e)
        {
          // TODO Auto-generated catch block
          e.printStackTrace();
        }
        catch (RemoteException e)
        {
          // TODO Auto-generated catch block
          e.printStackTrace();
        }
        catch (NotBoundException e)
        {
          // TODO Auto-generated catch block
          e.printStackTrace();
        }
      }
    }
  }

  /**
   * Re-routing message according to the original destination set in the
   * message.
   * 
   * @throws InterruptedException
   * @throws NotBoundException
   * @throws RemoteException
   * @throws AccessException
   */
  private synchronized void routingMessages(final InterNodeMessage message)
      throws AccessException, RemoteException, NotBoundException, InterruptedException
  {
    try
    {
      if (message.getMy_originaldest_node() == this.my_node_id)
      {
        if (message.getMethod_name() == REBALANCE)
        {
          Logger.out(this.my_node_id + " rebalance message from node " +
                     message.getMy_originalsource_node() + "received.");
          nodeman.rebalance();
        }
        else if (message.getMethod_name() == MULTIPLESTORE)
        {
          Logger.out(this.my_node_id +
                     " Fault tolerance storage message received from node - " +
                     message.getMy_originalsource_node());
          this.nodeman.duplicatestores.receivemessagesuccessor(message);
        }
        else if (message.getMethod_name() == FAULT_TOLERANCE)
        {
          final HandleRebalanceMessages rebhandler = new HandleRebalanceMessages(this.nodeman);
          rebhandler.faultTolhandler(message);
        }
        else if (message.getMethod_name() == PUT_KEY || message.getMethod_name() == GET_KEY ||
                 message.getMethod_name() == REMOVE_KEY)
        {
          processMessage(message);
        }
      }
      else
      {
        message.setMy_source_node(this.my_node_id);
        int dest_to_send =
            nodeman.hashhelp.optimalNodeFind(this.my_node_id,
                                             message.getMy_originaldest_node());
        message.setMy_dest_node(dest_to_send);
        String dest_node_str = nodeman.rmi_NodeFind(dest_to_send);
        final RemoteDistributedHashMap to_send =
            (RemoteDistributedHashMap) NodeManager.registry.lookup(dest_node_str);
        while (to_send.isrebalancing())
        {
          //Logger.out(" rebalancing");
        }
        to_send.send(message);
      }
    }
    catch (final RemoteException e)
    {
      Logger.out(this.my_node_id + " remote exception in sending message " + e);
      RebalanceChecks reb =
          new RebalanceChecks(this.nodeman, this.my_node_id, message.getMy_dest_node());
      int predecessor = reb.findPredecessor(message.getMy_dest_node());
      if (this.nodeman.rmi_NodeFind(message.getMy_dest_node()).endsWith("V"))
      {
        predecessor = reb.findPredecessor(predecessor);
      }
      Logger.out(" the pred is " + predecessor);
      reb.doallchecks();
      this.handlemessage(message, predecessor);
    }
    catch (final InterruptedException e)
    {
      Logger.writeContent(nodeman.LOGGER_FILE_NAME, " interrupted while sending");
      System.exit(1);
    }
    catch (final NotBoundException e)
    {
      System.exit(1);
    }
    catch (final AlreadyBoundException e)
    {
      Logger.out(" already bound exception in thread " + e);
      System.exit(1);
    }
  }

  /**
   * This method handles the message that caused to throw re-balancing.
   * @throws NotBoundException 
   * @throws RemoteException 
   * @throws AccessException 
   * @throws InterruptedException 
   */
  public void handlemessage(InterNodeMessage message ,int predecessor) 
      throws AccessException, RemoteException, NotBoundException, InterruptedException
  {
    
    final long time = System.currentTimeMillis();
    long newtime = System.currentTimeMillis();
    while (newtime - time < WAIT_MESSAGE)
    {
      newtime = System.currentTimeMillis();
    }
    String dest_node_str = this.nodeman.rmi_NodeFind(predecessor);
    RemoteDistributedHashMap to_send =
        (RemoteDistributedHashMap) NodeManager.registry.lookup(dest_node_str);
    while (to_send.isrebalancing())
    {
      //Logger.out(" rebalancing");
    }
    dest_node_str = this.nodeman.rmi_NodeFind(message.getMy_dest_node());
    to_send = (RemoteDistributedHashMap) NodeManager.registry.lookup(dest_node_str);
    Logger.out(this.my_node_id + " Sending message- " +
          message.getHashTableMessage().getkey() + " to node- " + dest_node_str);
    to_send.send(message);
  }

  /**
   * @throws AlreadyBoundException *
   * 
   */

  public synchronized void processMessagenode(final InterNodeMessage my_message)
      throws AlreadyBoundException
  {
    try
    {
      System.out.println(" inside node " + my_node_id);
      int destNode;
      String dest_str = null;

      if (!my_message.isAnswer_set() && my_message.getMy_originaldest_node() == my_node_id)
      {
        processRemoveMessage(my_message);
      }
      else
      {
        if (!my_message.isAnswer_set() && my_message.getMy_originaldest_node() != my_node_id)
        {
          // message not processed and on its way to the processing node.
          my_message.setMy_source_node(my_node_id);
          destNode =
              nodeman.hashhelp.optimalNodeFind(my_node_id,
                                               my_message.getMy_originaldest_node());
          dest_str = nodeman.rmi_NodeFind(destNode);
          final RemoteDistributedHashMap to_send =
              (RemoteDistributedHashMap) NodeManager.registry.lookup(dest_str);
          to_send.send(my_message);
        }
        if (my_message.isAnswer_set() && my_message.getMy_originaldest_node() != my_node_id)
        {
          System.out.println("Node Removal: message processed and on its way to dest " +
                             my_node_id);
          // message processed on its way to destination
          my_message.setMy_source_node(my_node_id);
          destNode =
              nodeman.hashhelp.optimalNodeFind(my_node_id,
                                               my_message.getMy_originaldest_node());
          dest_str = nodeman.rmi_NodeFind(destNode);
          if (destNode == my_message.getMy_originaldest_node())
          {
            System.out.println(" dest to send " + destNode);
            my_message.setMy_dest_node(destNode);
            dest_str = destNode + "A";
            System.out.println(" dest node " + dest_str);
          }
          final RemoteDistributedHashMap to_send =
              (RemoteDistributedHashMap) NodeManager.registry.lookup(dest_str);
          to_send.send(my_message);
        }
      }
    }
    catch (final RemoteException e)
    {
      Logger.writeContent(nodeman.LOGGER_FILE_NAME,
                          " remote exception in binding with registry/lookups " + e);
      Logger.out(this.my_node_id + " some nodes not present, need to rebalance");
      System.exit(1);
    }
    catch (final NotBoundException e)
    {

    }
    catch (InterruptedException e)
    {
      // TODO Auto-generated catch block
      e.printStackTrace();
    }
  }

  /**
   * process hash look up messages here.
   * 
   * @throws NotBoundException
   * @throws RemoteException
   * @throws AlreadyBoundException
   * @throws InterruptedException
   */
  private synchronized void processMessage(final InterNodeMessage message)
      throws RemoteException, NotBoundException, AlreadyBoundException, InterruptedException
  {
    int dest_to_send;
    String dest_node_str;
    if (message.isAnswer_set())
    {
      Logger.out(this.my_node_id + " A hash map message- " +
                 message.getHashTableMessage().getkey() + " received after processing ");
      final FutureMessage mes =
          nodeman.node_stubs.get(this.my_node_id).my_futures.get(message.getHashTableMessage()
              .getUniqueId());
      mes.set(message.getHashTableMessage().getvalue());
      nodeman.node_stubs.get(this.my_node_id).my_futures.remove(message.getHashTableMessage().getUniqueId());
      Logger.out(this.my_node_id + " The appropriate future is updated ");

    }
    else
    {
      Logger.out(this.my_node_id + " A new hash map message - " +
                 message.getHashTableMessage().getkey() + " received to be processed.");
      if (message.getMethod_name() == GET_KEY)
      {
        final Serializable val =
            nodeman.node_stubs.get(this.my_node_id).my_map.get(message.getHashTableMessage()
                .getkey());
        message.getHashTableMessage().setvalue(val);
        Logger.out(this.my_node_id + " Get message - " +
                   message.getHashTableMessage().getkey() + " processed.");
      }
      if (message.getMethod_name() == PUT_KEY)
      {
        Serializable val = null;
        synchronized (this)
        {
          val =
              nodeman.node_stubs.get(this.my_node_id).my_map.put(message.getHashTableMessage()
                  .getkey(), message.getHashTableMessage().getvalue());
        }
        this.nodeman.duplicatestores.sendmessagepredecessor(message, this.my_node_id);
        message.getHashTableMessage().setvalue(val);
      }
      if (message.getMethod_name() == REMOVE_KEY)
      {
        Serializable val = null;
        synchronized (this)
        {
          val =
              nodeman.node_stubs.get(this.my_node_id).my_map.remove(message
                  .getHashTableMessage().getkey());
        }
        message.getHashTableMessage().setvalue(val);
        Logger.out(this.my_node_id + " Remove message - " +
                   message.getHashTableMessage().getkey() + " processed.");
        this.nodeman.duplicatestores.sendmessagepredecessor(message, this.my_node_id);
      }
      
      if (message.getMethod_name() == REMOVE_NODE)
      {
        System.out.println("Node Remove:In processMessage");
        processRemoveMessage(message);
        System.out.println("Node Removal:Entered the actual node that need to simulate");
      }
      message.setAnswer_set(true);
      message.setMy_originaldest_node(message.getMy_originalsource_node());
      message.setMy_source_node(this.my_node_id);
      message.setMy_originalsource_node(this.my_node_id);
      dest_to_send =
          nodeman.hashhelp.optimalNodeFind(this.my_node_id, message.getMy_originaldest_node());
      message.setMy_dest_node(dest_to_send);
      dest_node_str = nodeman.rmi_NodeFind(dest_to_send);
      Logger.out(this.my_node_id + " messagage - " + message.getHashTableMessage().getkey() +
                 " processed and sending back to node " + message.getMy_originaldest_node());
      final RemoteDistributedHashMap to_send =
          (RemoteDistributedHashMap) NodeManager.registry.lookup(dest_node_str);
      while (to_send.isrebalancing())
      {
        //Logger.out(" rebalancing");
      }
      to_send.send(message);
    }
  }

  /**
   * Process the remove node's message.
   * 
   * @param the_message : stores the inter node message
   * @throws NotBoundException if the registry is not bound
   * @throws RemoteException remote exception
   * @throws AccessException if it cannot be accessed
   * @throws AlreadyBoundException
   */
  private synchronized void processRemoveMessage(final InterNodeMessage the_message)
      throws AccessException, RemoteException, NotBoundException, AlreadyBoundException
  {
    System.out.println("Node Remove:In processRemoveMessage");
    int dest_to_send;
    String dest_node_str;
    final RemoveNodeMessage internal_message = the_message.getRemoveNodeMessage();
    final ArrayList<Integer> node_list = internal_message.getRemoveNodesList();
    // final ArrayList<RemoveNodeChannel> message_channel =
    // internal_message.getNodeMessageChannelList();
    final ArrayList<HashMap<String, Serializable>> node_data =
        internal_message.getEachNodeData();
    // Ask the nodeManager to simulate these nodes and register back.
    // Update it and rebind it back.
    // Send the message to client that it is successfull
    for (int index = 0; index < node_list.size(); index++)
    {
      this.createVirtualNodes(node_list.get(index), node_data.get(index));

      // Access to remoteObjectCreation is required since it binds/ take care of
      // rebind
      // and also update actual and virtual info node list
    }
    NodeManager.registry.unbind(internal_message.getMyActualNode() + "A");
    System.out.println(" list of nodes in registry are " +
                       Arrays.toString(NodeManager.registry.list()));
  }

  private void createVirtualNodes(int the_node_no, HashMap<String, Serializable> the_data)
      throws RemoteException, NotBoundException, AlreadyBoundException
  {
    System.out.println("Node Removal:started the binding");
    nodeman.remoteArray[the_node_no] =
        new UnicastRemoteDistributedHashMap(the_node_no, false, nodeman.actual_node_number,
                                            this.nodeman);
    System.out.println(" create stuns ");
    nodeman.node_stubs.put(the_node_no, nodeman.remoteArray[the_node_no]);
    nodeman.remoteArray[the_node_no].setData(the_data);
    /////
    nodeman.node_stubs.get(the_node_no).my_map = the_data;
    /////
    nodeman.node_set.add(the_node_no);
    final RemoteDistributedHashMap stub =
        (RemoteDistributedHashMap) UnicastRemoteObject
            .exportObject(nodeman.remoteArray[the_node_no], 0);
    nodeman.messageChannelSetup(1);
    nodeman.QueueThreadStart(1);
    // set the channels and start thread.

    System.out.println(" exporting stubs ");
    final String rmi_name = Integer.toString(the_node_no).toString() + "V";
    System.out.println(" rmi name " + rmi_name);
    NodeManager.registry.rebind(rmi_name, stub);
    System.out.println(" binded ");
    System.out.println("Node Removal:done properly and rebinded ");
  }
}
